package xyz.heyaoshare.utils;

import cn.hutool.json.JSONUtil;
import io.netty.channel.ChannelOption;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.MediaType;
import org.springframework.http.client.reactive.ReactorClientHttpConnector;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.function.BodyInserters;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.reactive.function.client.WebClientResponseException;
import reactor.core.publisher.Flux;
import reactor.core.publisher.SignalType;
import reactor.netty.http.client.HttpClient;
import reactor.netty.resources.ConnectionProvider;
import xyz.heyaoshare.config.properties.WebProperties;

import java.io.IOException;
import java.time.Duration;

/**
 * Flux流工具类
 * @author YueHe
 * @version 0.0.1
 * @since 2025/6/25 上午10:50
 */
@Slf4j
@Component
public class StreamUtils {

    @Resource
    private WebProperties webProperties;

    /**
     * 使用WebClient以POST方式发送请求，并将响应转换为Flux<String>
     *
     * @param body 请求体内容，将被转换为JSON格式发送
     * @param url  请求的基础URL
     * @return 包含响应内容的Flux<String>，如果发生WebClientResponseException，则返回错误响应体内容
     */
    public Flux<String> simpleRequest(Object body, String url) {
        // 创建 HttpClient 并配置连接超时和响应超时
        HttpClient httpClient = HttpClient.create(getConnectionProvider())
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, webProperties.getStreamingConnectTimeout()) // 设置连接超时时间为 60 秒
                .responseTimeout(Duration.ofMillis(webProperties.getStreamingResponseTime())); // 设置响应超时时间为 180 秒

        // 构建 WebClient 实例
        WebClient webClient = WebClient.builder()
                .defaultHeader("Content-Type", MediaType.APPLICATION_JSON_VALUE) // 设置默认请求头为 JSON
                .baseUrl(url) // 设置请求的基础 URL
                .clientConnector(new ReactorClientHttpConnector(httpClient)) // 使用自定义的 HttpClient
                .build();

        // 发送请求并获取响应
        return webClient
                .post() // 配置为 POST 请求
                .accept(MediaType.TEXT_EVENT_STREAM) // 设置接受的响应类型为 text/event-stream
                .body(BodyInserters.fromValue(body)) // 将请求体设置为 JSON 格式的 body
                .retrieve() // 发送请求并获取响应
                .bodyToFlux(Object.class) // 将响应体转换为 Flux<Object>
                .map(this::mapHandler)// 将每个响应对象转换为 JSON 字符串
                .onErrorResume(this::errorResumeHandler) // 捕获并处理异常
                .doOnError(this::onErrorHandler) // 记录未被捕获的错误
                .doFinally(this::finallyHandler)// 在流结束时执行清理操作
                .startWith("Chat started: Waiting for the first message.") // 在流开始时插入自定义消息
                .concatWithValues("Chat ended: All messages have been received."); // 在流结束时插入自定义消息
    }

    /**
     * 获取自定义的连接提供器
     *
     * @return 返回配置好的连接提供器
     */
    private ConnectionProvider getConnectionProvider() {
        return ConnectionProvider.builder("custom")
                .maxConnections(100) // 设置最大连接数为 100
                .maxIdleTime(Duration.ofSeconds(30)) // 设置连接最大闲置时间为 30 秒
                .pendingAcquireTimeout(Duration.ofMillis(60000)) // 设置获取连接的超时时间为 100 毫秒
                .build();
    }

    /**
     * 将传入的Object对象转换为JSON字符串并返回
     *
     * @param s 待转换的Object对象
     * @return 转换后的JSON字符串
     */
    private String mapHandler(Object s) {
        log.info("Received message: {}\n", JSONUtil.toJsonStr(s)); // 记录接收到的消息
        return JSONUtil.toJsonStr(s);
    }

    /**
     * 根据信号类型进行最终处理
     *
     * @param signalType 信号类型
     */
    private void finallyHandler(SignalType signalType) {
        switch (signalType) {
            case ON_COMPLETE -> log.info("处理完成..."); // 流正常完成
            case CANCEL -> log.info("客户端主动断开连接..."); // 客户端取消了流
            case ON_ERROR -> log.error("响应错误终止..."); // 流因错误终止
        }
    }

    /**
     * 错误处理函数
     *
     * @param e 异常对象
     */
    private void onErrorHandler(Throwable e) {
        log.error("Error occurred: {}", e.getMessage());
    }

    /**
     * 错误恢复处理器
     *
     * @param e 异常对象
     * @return 返回字符串流，如果异常无法处理则返回错误流
     */
    private Flux<String> errorResumeHandler(Throwable e) {
        if (e instanceof WebClientResponseException w) { // 如果是 WebClientResponseException
            log.error("WebClientResponseException occurred: {}", w.getMessage());// 记录错误日志
            return Flux.just(w.getResponseBodyAsString());
        }
        if (e instanceof IOException i) { // 如果是 IOException
            if (i.getMessage().contains("Broken pipe")) { // 检查是否是 Broken pipe 异常
                log.error("Broken pipe exception occurred: {}", i.getMessage()); // 记录错误日志
                return Flux.just("Connection closed by client.");
            }
        }
        return Flux.error(e);
    }

}